package com.hzlj.position.job.listener;

import com.fz.common.kafka.annotation.KafkaListenerRetry;
import com.google.common.collect.Lists;
import com.hzlj.position.config.config.PositionConfig;
import com.hzlj.position.locate.common.model.PositionSkynet;
import com.hzlj.position.locate.service.PositionSkynetService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

import static com.hzlj.position.config.common.constant.Topic.POSITION_SKYNET_ADD;


/**
 * @author lifh
 * @date 2023/3/17 11:29
 **/
@Slf4j
@Component
public class PositionSkynetAddListener {
    @Resource
    private PositionSkynetService positionSkynetService;
    @Resource
    private PositionConfig positionConfig;

    private List<PositionSkynet> positions = Lists.newArrayList();

    @KafkaListenerRetry(enable = false)
    @KafkaListener(topics = {POSITION_SKYNET_ADD})
    public void positionAdd(ConsumerRecord<String, PositionSkynet> record) {
        positions.add(record.value());
        boolean needAdd = positions.size() >= positionConfig.getAddBatchSize();
        try {
            if (needAdd) {
                positionSkynetService.savePositionSkynet(positions);
            }
        } catch (Exception e) {
            log.error("【kafka】【天网定位数据】批量新增异常,deviceCode-channel={}",
                    positions.stream().map(ee -> ee.getDeviceCode() + "-" + ee.getChannel()).collect(Collectors.joining(",")), e);
        } finally {
            if (needAdd) {
                positions.clear();
            }
        }
    }


}
